package com.baishancloud.log.fm

import com.baishancloud.log.common.util.{StringUtil, TimeUtil}
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.common.functions.MapFunction

/**
 * 接受的是二元组，第一个元素用来区分来源：LUBAN/BLB
 *
 * @author ziqiang.wang 
 * @date 2022/3/7 17:13
 */
class LogParse extends MapFunction[(String, Origin), Log] {

  override def map(value: (String, Origin)): Log = {
    val origin = value._2
    val split: Array[String] = origin.complexField1().split("@")
    val upstreamStatusStr: String = if (split.length < 2) "0" else split(1)
    val upstreamStatus: Int =
      try {
        if (upstreamStatusStr.contains(",")) {
          val split: Array[String] = upstreamStatusStr.split(",")
          split(split.length - 1).toInt
        } else if (StringUtil.isLong(upstreamStatusStr)) {
          upstreamStatusStr.toInt
        } else {
          0
        }
      } catch {
        case _: Exception => 0
      }

    Log(TimeUtil.timeAtLong(origin.datetimeOfUTC().toEpochSecond, 60),
      origin.host(),
      origin.httpCode(),
      if (origin.hit().endsWith("MISS")) 0.toByte else 1.toByte,
      origin.domain(),
      value._1,
      if (origin.ip().contains(".")) 1 else 0,
      upstreamStatus,
      1L,
      0L,
      origin.traffic(),
      origin.responseTime())
  }
}
